package cn.scw.framework.protocol;

import cn.scw.common.utils.spring.SpringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 处理器管理器，是数据处理的入口
 */
@Component
public class DataResolverManager {

    private Logger logger= LoggerFactory.getLogger(this.getClass());

    private static final String START_RESOLVER_NAME="DispatchResolver";
    //待解析的消息的队列最大长度
    private static final int QUEUE_MAX_LENGTH=5000;
    //处理器线程数量
    private static final int THREAD_POOL_SIZE=16;

    public static DataResolverManager getInstance(){
        return SpringUtils.getBean(DataResolverManager.class);
    }

    //所有的DataResolver都在这里集中进行管理
    private Map<String,DataResolver> dataResolvers =new HashMap<>();
    //带解析的消息的队列
    private BlockingQueue<PendingData> dataQueue=new ArrayBlockingQueue<>(QUEUE_MAX_LENGTH);
    private ExecutorService executorService;

    /**
     * 管理器初始化方法
     */
    public void init(){

        //初始化线程池
        executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);

        for(int i = 0; i<THREAD_POOL_SIZE; i++) {

            executorService.submit((Runnable) () -> {

                DataResolver startResolver=SpringUtils.getBean(BaseDataResolver.class);
                if(startResolver==null){
                    throw new RuntimeException("未获取到起始处理器："+START_RESOLVER_NAME+"的实例");
                }

                while(true) {
                    try {
                        //从队列中获取数据
                        PendingData pendingData = null;
                        try {
                            pendingData = dataQueue.take();
                        } catch (InterruptedException e) {
                            logger.error("获取数据失败");
                        }
                        //如果获取到数据则处理数据
                        if (pendingData != null) {
                            startResolver.resolveData(null,pendingData.getData(), pendingData.getContext());
                        } else {
                            //否则睡眠3秒再监听
                            try {
                                Thread.sleep(3000L);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    }catch (Exception ex){
                        logger.error(ex.getMessage());
                    }
                }
            });
        }

//        executorService.shutdown();
    }

    /**
     * 处理数据
     * @param data 需要被处理的数据
     */
    public void resolveData(byte[] data, ResolverContext context){

        PendingData pendingData=new PendingData();
        pendingData.setData(data);
        pendingData.setContext(context);

        try {
            this.dataQueue.put(pendingData);
        } catch (InterruptedException e) {
            logger.error("消息插入失败。"+e.getMessage());
        }
    }
}
